
#include "QueryPlan.h"

using namespace std;

extern struct FuncOperator *finalFunction;
extern struct TableList *tables;
extern struct AndList *boolean;
extern struct NameList *groupingAtts;
extern struct NameList *attsToSelect;
extern int distinctAtts;
extern struct FuncOperator *finalfunc;
extern FILE *yyin;

extern "C" {
	int yyparse(void);   // defined in y.tab.c
	int yyfuncparse(void);   // defined in yyfunc.tab.c
	struct YY_BUFFER_STATE *yy_scan_string(const char*);
}


void QueryPlan::LoadStats(char *fName, char *relNames[], int numTables){
        FILE *foo = fopen (fName, "r");

        // this is enough space to hold any tokens
        char space[200];

        fscanf (foo, "%s", space);
        int totscans = 1;

        // see if the file starts with the correct keyword
        if (strcmp (space, "BEGIN")) {
                cout << "Unfortunately, this does not seem to be a schema file.\n";
                exit (1);
        }

	bool *processed = new bool[numTables];
	for (int i = 0; i < numTables; i++)
		processed[i] = false;
	int prCount = 0;

        while (1)
	{
                // check to see if this is the one we want
                fscanf (foo, "%s", space);
                totscans++;
		bool fl = false;
		int i = 0;
		for ( ; i < numTables; i++)
		{
			if (processed[i])
				continue;
                	if (!strcmp (space, relNames[i]))
			{
				fl = true;
				break;
			}
		}

		if (fl == false)
		{
                       	// it is not, so suck up everything to past the BEGIN
	                while (1) {

      	                	// suck up another token
               	        	if (fscanf (foo, "%s", space) == EOF) {
                       	                cerr << "Could not find the schema for the specified relation.\n";
                               	        exit (1);
                                }

       	                        totscans++;
               	                if (!strcmp (space, "BEGIN")) {
                       	                break;
                               	}
                       	}

	                // otherwise, got the correct file!!
        	}
		else
		{
        		char Att[200], cnt[200];
			int count;
			fscanf (foo, "%s", cnt);
			count = atoi(cnt);
			stats.AddRel(space, count);

        		// count the number of attributes specified
        		while (1) {
		                fscanf (foo, "%s", Att);
		                if (!strcmp (Att, "END")) {
                       			break;
			        }
				else
				{
		                        fscanf (foo, "%s", cnt);
					count = atoi(cnt);
					stats.AddAtt(space, Att, count);
                		}
			}
			processed[i] = true;
			prCount++;
			if (prCount == numTables)
			{
				break;
			}
			if (fscanf (foo, "%s", space) == EOF) {
                               cerr << "Could not find the schema for the specified relation.\n";
                               return;
                        }

        	}
	}

        // now actually load up the schema
        fclose (foo);
}

void QueryPlan::Setup (char *tables[], char *aliases[], int numTables) {
	FILE *fp = fopen (settings, "r");
	if (fp) {
		char *mem = (char *) malloc (80 * 4);
		catalog_path = &mem[0];
		dbfile_dir = &mem[80];
		tpch_dir = &mem[160];
		stats_path = &mem[240];
		char line[80];
		fgets (line, 80, fp);
		sscanf (line, "%s\n", catalog_path);
		fgets (line, 80, fp);
		sscanf (line, "%s\n", dbfile_dir);
		fgets (line, 80, fp);
		sscanf (line, "%s\n", tpch_dir);
		fgets (line, 80, fp);
		sscanf (line, "%s\n", stats_path);
		fclose (fp);
		if (! (catalog_path && dbfile_dir && tpch_dir)) {
			cerr << " Test settings file 'test.cat' not in correct format.\n";
			free (mem);
			exit (1);
		}
	}
	else {
		cerr << " Test settings files 'test.cat' missing \n";
		exit (1);
	}

	for (int i = 0; i < 10; i++)
	{
		R[i] = NULL;
	}

	for (int i = 0; i < numTables; i++)
	{
		R[i] = new relation (tables[i], new Schema (catalog_path, tables[i]), dbfile_dir);
		if (aliases[i] != NULL)
			R1[i] = new relation (tables[i], new Schema (catalog_path, tables[i], aliases[i]), dbfile_dir);
		else
			R1[i] = NULL;
	}
}

QueryPlan::QueryPlan(int opt) {

	struct TableList *tables1;
	tables1 = tables;
	char *relNames[20];
	char *aliasNames[20];
	int relCnt = 0;
	while (tables1 != NULL)
	{
		relNames[relCnt] = tables1->tableName;
		aliasNames[relCnt] = tables1->aliasAs;
		relCnt++;
		tables1 = tables1->next;
	}

	Setup(relNames, aliasNames, relCnt);
	LoadStats(stats_path, relNames, relCnt);

	for (int i = 0; i < relCnt; i++)
	{
		tblsSchema.insert(pair<string, relation *>(relNames[i], R[i]));
		if (aliasNames[i] != NULL)
		{
			tblsSchema.insert(pair<string, relation *>(aliasNames[i], R1[i]));
			stats.CopyRel(relNames[i], aliasNames[i]);
		}
	}
}

QueryPlan::~QueryPlan() {

}

void QueryPlan::ProcessAndList() {

	struct TableList *tables1;
	struct AndList *boolean1;
	tables1 = tables;
	boolean1 = boolean;

	if (tables1 != NULL) {
		while (tables1 != NULL) {
			tableSets.insert(tables1->tableName);
			tableSets.insert(tables1->aliasAs);
			tables1 = tables1->next;
		}
		char *relNames[20];
		set<string>::iterator it;
		int i=0;
		for (it = tableSets.begin(); it != tableSets.end(); it++) {
			string str = *it;
			char * tmp = new char[str.size() + 1];
			std::copy(str.begin(), str.end(), tmp);
			relNames[i++] = tmp;
		}
		if (boolean1 != NULL) {
			while (boolean1 != NULL) {
				struct OrList *pOr = boolean1->left;
				struct AndList *tAndList = new AndList();
				if (pOr != NULL) {
					struct ComparisonOp *pCom = pOr->left;
					if (pCom != NULL) {
						struct Operand *pOperandL = NULL;
						struct Operand *pOperandR = NULL;

						pOperandL = pCom->left;
						string lRelName;
						string attLeft = pOperandL->value;
						if(!attLeft.empty()){
                            char *tmpL = new char[attLeft.size() + 1];
                            std::copy(attLeft.begin(), attLeft.end(), tmpL);
                            RelData lRel = stats.findAtt(attLeft, relNames, tableSets.size());
                            lRelName = lRel.name;
						}


						pOperandR = pCom->right;
						string rRelName;
						string attRight = pOperandR->value;
						if(!attRight.empty()){
                            char *tmpR = new char[attRight.size() + 1];
                            std::copy(attRight.begin(), attRight.end(), tmpR);
                            RelData rRel = stats.findAtt(attRight, relNames, tableSets.size());
                            rRelName = rRel.name;
						}

                        if ((lRelName.empty() && rRelName.empty()) || (!lRelName.empty() && !rRelName.empty())) {
							if(lRelName != rRelName && pCom->code == 7){
								//This is a join
								pair<string, struct OrList*> jpairL (lRelName+","+rRelName, pOr);
								pair<string,struct OrList*> jpairR (rRelName+","+lRelName, pOr);

								joinAtts.insert(jpairL);
								joinAtts.insert(jpairR);
								joinTables.insert(lRelName);
								joinTables.insert(rRelName);
							}
						}
						else if (!lRelName.empty() && rRelName.empty()) {
							map<string,AndList*>::iterator it = tableCNFs.find(lRelName);
							set<string> tbls;
							struct OrList *pOrTmp = pOr;
							while (pOrTmp->rightOr)
							{
								RelData rRel = stats.findAtt(pOrTmp->left->left->value, relNames, tableSets.size());
								if (rRel.name.empty())
								{
									RelData rRel = stats.findAtt(pOrTmp->left->right->value, relNames, tableSets.size());
									if (rRel.name.empty())
									{
										cerr<<"Both the operands are not found in the given relations\n";
										exit(1);
									}
									tbls.insert(rRel.name);
								}
								else
								{
									tbls.insert(rRel.name);
								}
								pOrTmp = pOrTmp->rightOr;
							}
							if(it!=tableCNFs.end()){
								tAndList = it->second;
								struct AndList *tempAndList = new AndList();
								tempAndList->left = pOr;
								tempAndList->rightAnd = NULL;
								tAndList->rightAnd = tempAndList;
							}
							else
							{
								struct AndList *tempAndList = new AndList();
								tempAndList->left = pOr;
								tempAndList->rightAnd = NULL;
								pair<string,AndList*> apair (lRelName, tempAndList);
								tableCNFs.insert(apair);
							}
						}
						if (lRelName.empty() && !rRelName.empty()) {
							map<string, AndList*>::iterator it = tableCNFs.find(rRelName);
							if (it != tableCNFs.end()) {
								tAndList = it->second;
								struct AndList *tempAndList = new AndList();
								tempAndList->left = pOr;
								tempAndList->rightAnd = NULL;
								tAndList->rightAnd = tempAndList;
							} else {
								struct AndList *tempAndList = new AndList();
								tempAndList->left = pOr;
								tempAndList->rightAnd = NULL;
								pair<string, AndList*> apair(rRelName, tempAndList);
								tableCNFs.insert(apair);
							}
						}
					}
				}
				boolean1 = boolean1->rightAnd;
			}
		}
	}
}

void QueryPlan::Print(QueryPlanNode *p){
	if (p == NULL)
		return;
	Print(p->left);
	p->Print();
	Print(p->right);

}

void QueryPlan::ExecuteTree(QueryPlanNode *p){
	if (p == NULL)
		return;
	ExecuteTree(p->left);
	ExecuteTree(p->right);
	p->Execute();
}

void QueryPlan::Clean(QueryPlanNode *p){
	if (p == NULL)
		return;
	Clean(p->left);
	Clean(p->right);
	p->Clean();
	delete p;
}

void QueryPlan::WaitUntilDone(QueryPlanNode *p){
	if (p == NULL)
		return;
	WaitUntilDone(p->left);
	WaitUntilDone(p->right);
	p->Wait();
}

QueryPlanNode* QueryPlan::Planner(){
	char *relNames[20];
	int  relCount = 0;
	int pipeId = 1;
	set<string>::iterator i;
	double best = numeric_limits<double>::max();
	int best_i;
	int best_j;
	struct AndList best_cnf;
	Attribute joinatts[1024];
	int attCount = 0;
	double result;

	if (joinTables.size() <= 1)
		return NULL;

	vector<string> myvector;
	for (i = joinTables.begin() ; i != joinTables.end(); i++)
		myvector.push_back(*i);

	for (int x = 0;  x < 20; x++)
		relNames[x] = (char *)malloc(100);

    int joinType = 0;
    int vSize;
    FILE *fileJoinType = fopen("joinType", "r");
    if (fileJoinType) {
        fscanf(fileJoinType, "%d", &joinType);
    }
    fclose(fileJoinType);

    //for sort
    if (joinType == 1){
        vSize = myvector.size();
        for (int i1 = 0; i1 < vSize; i1++)
        {
            for (int j1 = i1 + 1; j1 < vSize; j1++)
            {
                map<string, struct OrList*>::iterator k;
                k = joinAtts.find(myvector[i1] + "," + myvector[j1]);
                strcpy (relNames[0], (char *)myvector[i1].c_str());
                strcpy (relNames[1], (char *)myvector[j1].c_str());

                if (k == joinAtts.end())
                {
                    result = stats.Estimate(NULL, relNames, 2);
                }
                else
                {
                    struct AndList andList;
                    andList.left = k->second;
                    andList.rightAnd = NULL;
                    result = stats.Estimate(&andList, relNames, 2);
                }
                if (result < best)
                {
                    if (k == joinAtts.end())
                        best_cnf = {NULL, NULL};
                    else
                        best_cnf = {k->second, NULL};
                    best_i = i1;
                    best_j = j1;
                    best = result;
                }
            }
        }
    }

    //for hash
    else
    {
        vSize = myvector.size();
        int max = 0;
        int maxRelation = 0;
        for (int i = 0; i < vSize; i++)
        {
            strcpy (relNames[0], (char *)myvector[i].c_str());
            int temp = stats.Estimate(NULL, relNames, 1);
            if (temp > max){
                max = temp;
                maxRelation = i;
            }
        }

        for (int j = 0; j < vSize; j++)
        {
            map<string, struct OrList*>::iterator k;
            k = joinAtts.find(myvector[maxRelation] + "," + myvector[j]);
            strcpy (relNames[0], (char *)myvector[maxRelation].c_str());
            strcpy (relNames[1], (char *)myvector[j].c_str());

            if (k == joinAtts.end())
            {
                result = stats.Estimate(NULL, relNames, 2);
            }
            else
            {
                struct AndList andList;
                andList.left = k->second;
                andList.rightAnd = NULL;
                result = stats.Estimate(&andList, relNames, 2);
            }
            if (result < best)
            {
                if (k == joinAtts.end())
                    best_cnf = {NULL, NULL};
                else
                    best_cnf = {k->second, NULL};
                best_i = maxRelation;
                best_j = j;
                best = result;
            }
        }
    }

    strcpy(relNames[0], (char *)myvector[best_i].c_str());
	strcpy(relNames[1], (char *)myvector[best_j].c_str());

	relCount = 2;
	if (best_cnf.left == NULL && best_cnf.rightAnd == NULL)
	{
		stats.Apply(NULL, relNames, 2);
	}
	else
	{
		stats.Apply(&best_cnf, relNames, 2);
	}

	int inPipeIdL, inPipeIdR, outPipeId;
	Schema *schemaL, *schemaR, *outSchema;
	Record *literal = new Record();
	CNF *cnf = new CNF();
	relation *r, *l;
	map<string, relation *>::iterator it;
	it = tblsSchema.find(myvector[best_i]);
	l = it->second;
	it = tblsSchema.find(myvector[best_j]);
	r = it->second;
	schemaL = l->schema();
	schemaR = r->schema();
	int lnumAtts = schemaL->GetNumAtts();
	int rnumAtts = schemaR->GetNumAtts();
	Attribute *lAtts = schemaL->GetAtts();
	Attribute *rAtts = schemaR->GetAtts();
	for (int ii = 0; ii < lnumAtts; ii++)
	{
		joinatts[attCount].myType = lAtts[ii].myType;
		joinatts[attCount].name = strdup (lAtts[ii].name);
		attCount++;
	}
	for (int ii = 0; ii < rnumAtts; ii++)
	{
		joinatts[attCount].myType = rAtts[ii].myType;
		joinatts[attCount].name = strdup (rAtts[ii].name);
		attCount++;
	}

	cnf->GrowFromParseTree (&best_cnf, schemaL, schemaR, *literal);
	Pipe *lPipe = new Pipe(RUN_LENGTH);
	Pipe *rPipe = new Pipe(RUN_LENGTH);
	Pipe *oPipe = new Pipe(RUN_LENGTH);
	QueryPlanNode * ptr;
    ptr = new JoinNode(pipeId, pipeId + 1, pipeId + 2, cnf, literal, schemaL, schemaR, lPipe, rPipe, oPipe);
	ptr->node_out_sch = new Schema ("join sch", attCount, joinatts);
	ptr->out_pipe = oPipe;
	map<string, AndList *>::iterator mit;
	mit = tableCNFs.find(myvector[best_i]);
	DBFile *file = new DBFile();
	file->Open(l->path());
	cnf = new CNF();
	literal = new Record();

	cnf->GrowFromParseTree(mit->second, l->schema(), *literal);
	QueryPlanNode * left = new SelectFileNode(file, (char *)myvector[best_i].c_str(), pipeId, cnf, literal, lPipe);
	file = new DBFile();
	file->Open(r->path());
	cnf = new CNF();
	literal = new Record();
	mit = tableCNFs.find(myvector[best_j]);
	cnf->GrowFromParseTree(mit->second, r->schema(), *literal);
	QueryPlanNode * right = new SelectFileNode(file, (char *)myvector[best_j].c_str(), pipeId + 1, cnf, literal, rPipe);
	pipeId = 3;
	ptr->left = left;
	ptr->right = right;

	if ( best_i > best_j)
	{
		myvector.erase(myvector.begin() + best_i);
		myvector.erase(myvector.begin() + best_j);
	}
	else
	{
		myvector.erase(myvector.begin() + best_j);
		myvector.erase(myvector.begin() + best_i);
	}

	while (myvector.size() > 0)
	{
		best = numeric_limits<double>::max();
		set<string>::iterator j;
		vSize = myvector.size();
		for (int j1 = 0; j1 < vSize; j1++)
		{
			map<string, OrList*>::iterator k;
			double result;
			int l = 0;
			for ( ; l < relCount; l++)
			{
				string temp = relNames[l];
				string sss = temp + "," + myvector[j1];
				k = joinAtts.find(temp + "," + myvector[j1]);
				if (k != joinAtts.end())
					break;
			}

			strcpy(relNames[relCount], (char *)myvector[j1].c_str());

			if (l == relCount)
				result = stats.Estimate(NULL, relNames, relCount + 1);
			else
			{
				struct AndList andList;
				andList.left = k->second;
				andList.rightAnd = NULL;
				result = stats.Estimate(&andList, relNames, relCount + 1);
			}
			if (result < best)
			{
			    if (l == relCount)
                    best_cnf = {NULL, NULL};
                else
                    best_cnf = {k->second, NULL};

				best_j = j1;
				best = result;
			}
		}

		strcpy(relNames[relCount++], (char *)myvector[best_j].c_str());
		if (best_cnf.left == NULL && best_cnf.rightAnd == NULL){
			stats.Apply(NULL, relNames, relCount);
		}
		else{
			stats.Apply(&best_cnf, relNames, relCount);
		}

        literal = new Record();
        cnf = new CNF();
        map<string, relation *>::iterator it;
        it = tblsSchema.find(myvector[best_j]);
        r = it->second;

        schemaR = r->schema();
        rnumAtts = schemaR->GetNumAtts();
        rAtts = schemaR->GetAtts();
        for (int ii = 0; ii < rnumAtts; ii++)
        {
            joinatts[attCount].myType = rAtts[ii].myType;
            joinatts[attCount].name = strdup (rAtts[ii].name);
            attCount++;
        }

		left = ptr;
		lPipe = oPipe;
		rPipe = new Pipe(RUN_LENGTH);
		oPipe = new Pipe(RUN_LENGTH);
		schemaL = left->node_out_sch;
		cnf->GrowFromParseTree (&best_cnf, schemaL, schemaR, *literal);
        ptr = new JoinNode(pipeId, pipeId + 1, pipeId + 2, cnf, literal, schemaL, schemaR, lPipe, rPipe, oPipe);
        ptr->node_out_sch = new Schema ("join sch", attCount, joinatts);
        file = new DBFile();
		file->Open(r->path());
		cnf = new CNF();
		literal = new Record();
		mit = tableCNFs.find(myvector[best_j]);
		cnf->GrowFromParseTree(mit->second, r->schema(), *literal);
		QueryPlanNode * right = new SelectFileNode(file, (char *)myvector[best_j].c_str(), pipeId + 1, cnf, literal, rPipe);
		pipeId += 2;
		ptr->left = left;
		ptr->right = right;

		ptr->out_pipe = oPipe;
        myvector.erase(myvector.begin() + best_j);
	}

	pipeCount = pipeId;

	return ptr;
}

QueryPlanNode* QueryPlan::GetOptimalPlan() {

	ProcessAndList();
	struct FuncOperator *finalFunction1;
	struct TableList *tables1;
	struct AndList *boolean1;
	struct NameList *groupingAtts1;
	struct NameList *attsToSelect1;
	int distinctAtts1;
	int distinctFunc1;
	finalFunction1 = finalFunction;
	tables1 = tables;
	boolean1 = boolean;
	groupingAtts1 = groupingAtts;
	attsToSelect1 = attsToSelect;

	QueryPlanNode *queryNode = NULL;
	QueryPlanNode *root = NULL;
	Schema *updatedSch;
	Pipe *iPipe;
	Pipe *oPipe;
	int pipeId = 1;
	pipeCount = 0;

	int tcnt = 0;
	if (tables1 != NULL) {
		while (tables1 != NULL) {
			tcnt++;
			tables1 = tables1->next;
		}
	}
	else{
		cout<<"No Tables in the Query";
		exit(1);
	}

    tables1 = tables;
	if (tcnt > joinTables.size()){
	    while (tables1 != NULL) {
            bool found = false;
	        for(set<string>::iterator it = joinTables.begin(); it != joinTables.end(); ++it){
	           if ((*it).compare(tables1->aliasAs) == 0){
                    found = true;
                    break;
	           }
            }

            if (!found)
                joinTables.insert(tables1->aliasAs);

			tables1 = tables1->next;
		}
	}

	if (tcnt > 1) {
		queryNode = Planner();
		updatedSch = queryNode->node_out_sch;
		iPipe = queryNode->out_pipe;
		root = queryNode;
	}
	else {
		DBFile *file = new DBFile();
		map<string, relation *>::iterator it;
		if(tables->aliasAs != NULL){
			it = tblsSchema.find(tables->aliasAs);
		}
		else{
			it = tblsSchema.find(tables->tableName);
		}
		r = it->second;
		CNF *cnf =  new CNF();
		oPipe = new Pipe(RUN_LENGTH);
		Record *literal = new Record();
		file->Open(r->path());
		cnf->GrowFromParseTree(boolean1, r->schema(), *literal);
		QueryPlanNode *node = new SelectFileNode(file, r->name(), pipeId, cnf, literal, oPipe);
		node->left = NULL;
		node->right = NULL;
		if (queryNode == NULL) {
			queryNode = node;
			root = node;
		}
		else {
			node->left = queryNode;
			queryNode = node;
		}
		iPipe = oPipe;
		updatedSch = r->schema();
	}

    if (pipeCount > pipeId)
        pipeId = pipeCount;

	//sum
	if (finalFunction1 != NULL && groupingAtts1 == NULL) {
		oPipe = new Pipe(RUN_LENGTH);

		Function *computeMe = new Function();
		computeMe->GrowFromParseTree(finalFunction1, *updatedSch);

		QueryPlanNode *node = new SumNode(pipeId, pipeId + 1, computeMe, iPipe, oPipe);

		node->left = NULL;
		node->right = NULL;
		if (queryNode == NULL) {
			queryNode = node;
			root = node;
		} else {
			node->left = queryNode;
			queryNode = node;
		}

		Schema *sum_sch;
		Attribute sum_att;
		if (computeMe->returnsInt) {
			sum_att = {"Sum", Int};
			updatedSch = new Schema("sum_sch", 1, &sum_att);
		}
		else
		{
			sum_att = {"Sum", Double};
			updatedSch = new Schema("sum_sch", 1, &sum_att);
		}

		iPipe = oPipe;
	}

	//group by
	if (groupingAtts1 != NULL && finalFunction1 != NULL) {
		Function *computeMe = new Function();
		oPipe = new Pipe(RUN_LENGTH);
		vector<string> groupingAttsList;

		while (groupingAtts1 != NULL) {
			groupingAttsList.push_back(groupingAtts1->name);
			groupingAtts1 = groupingAtts1->next;
		}

		computeMe->GrowFromParseTree(finalFunction1, *updatedSch);
		OrderMaker *order = new OrderMaker();
		int i = 0;

		for(int k = groupingAttsList.size() - 1; k >= 0; k--) {

			order->numAtts++;
			order->whichAtts[i] = updatedSch->Find((char*)groupingAttsList[k].c_str());
			order->whichTypes[i] = updatedSch->FindType((char*)groupingAttsList[k].c_str());
			i++;
		}
		QueryPlanNode *node = new GroupByNode(pipeId, pipeId + 1, order, computeMe, iPipe, oPipe);
		node->left = NULL;
		node->right = NULL;
		if (queryNode == NULL) {
			queryNode = node;
			root = node;
		} else {
			node->left = queryNode;
			queryNode = node;
		}

		Attribute sumGroupByAttrs[i+1];
		Attribute sum_att;
		if(computeMe->returnsInt)
		{
			 sum_att = {"Sum", Int};
		}
		else
		{
			sum_att = {"Sum", Double};
		}
		sumGroupByAttrs[0] = sum_att;
		int j = 1;

		for(int l = groupingAttsList.size() - 1; l >= 0; l--) {
			Attribute att;
			att.name = (char*)groupingAttsList[l].c_str();
			att.myType = updatedSch->FindType((char*)groupingAttsList[l].c_str());
			sumGroupByAttrs[j++] = att;
		}

		updatedSch = new Schema("sumGrpBySch", j, sumGroupByAttrs);
		iPipe = oPipe;
	}

	// project
	if (attsToSelect1 != NULL) {
		vector<string> attsToSelectList;

		while (attsToSelect1 != NULL) {
			attsToSelectList.push_back(attsToSelect1->name);
			attsToSelect1 = attsToSelect1->next;
		}

		vector<int> *keepMe = new vector<int>;
		int numAttsInput = updatedSch->GetNumAtts();
		int numAttsOutput = 0;
		oPipe = new Pipe(RUN_LENGTH);

		int sum_pos = updatedSch->Find("Sum");
		if(sum_pos != -1){
			numAttsOutput++;
			keepMe->push_back(sum_pos);
		}

		for(int k = attsToSelectList.size() - 1; k >= 0; k--) {
			keepMe->push_back(updatedSch->Find((char*)attsToSelectList[k].c_str()));
			numAttsOutput++;
		}

		QueryPlanNode *node = new ProjectNode(pipeId, pipeId + 1, keepMe, numAttsInput, numAttsOutput, iPipe, oPipe);
		node->left = NULL;
		node->right = NULL;
		if (queryNode == NULL) {
			queryNode = node;
			root = node;
		} else {
			node->left = queryNode;
			queryNode = node;
		}

		Attribute pAttrs[numAttsOutput];
		int j = 0;

		if(sum_pos != -1){
			Attribute att;
			att.name = "Sum";
			att.myType = updatedSch->FindType("Sum");
			pAttrs[j++] = att;
		}

		for(int l = attsToSelectList.size() - 1; l >= 0; l--) {
			Attribute att;
			att.name = (char*)attsToSelectList[l].c_str();
			att.myType = updatedSch->FindType((char*)attsToSelectList[l].c_str());
			pAttrs[j++] = att;
		}
		updatedSch = new Schema("p_sch", numAttsOutput, pAttrs);
		iPipe = oPipe;
	}

	// distinct
	if (distinctAtts > 0) {
		oPipe = new Pipe(RUN_LENGTH);
		QueryPlanNode *node = new DistinctNode(pipeId, pipeId + 1, updatedSch, iPipe, oPipe);
		node->left = NULL;
		node->right = NULL;
		if (queryNode == NULL) {
			queryNode = node;
			root = node;
		} else {
			node->left = queryNode;
			queryNode = node;
		}
	}
	root = queryNode;
	root->out_pipe = oPipe;
	root->node_out_sch = updatedSch;
	return root;

}

